package hadoop.classification.linear;

import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapred.FileInputFormat;
import org.apache.hadoop.mapred.FileOutputFormat;
import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.TextInputFormat;
import org.apache.hadoop.mapred.TextOutputFormat;

/**
 * Logistic Regression Invoker on Hadoop
 * @author Erheng Zhong (purlin.zhong@gmail.com)
 *
 */
public class HadoopLR {
	
	/**
	 * Model Building
	 * @param inputPath: training data path on hadoop
	 * @param outputPath: output path
	 * @param modelName: built model name
	 * @param numFeatures: number of features 
	 * @param numClasses: number of classes
	 * @param numIt: number of iterations
	 * @param learningRate: learning rate
	 * @param eps: threshold for termination 
	 * @param master: IP of spanning tree server
	 * @param masterPort: Port of spanning tree server
	 * @throws Exception
	 */
	public static void buildModel(String inputPath, String outputPath, String modelName, float d, int numFeatures, int numClasses, int numIt, float learningRate, float eps, String master, int masterPort) throws Exception {
		JobConf conf = new JobConf(HadoopLR.class);
		conf.setInt("data.numfeatures", numFeatures);
		conf.setInt("data.numclasses", numClasses);
		conf.setInt("model.numit", numIt);
		conf.setFloat("model.learningrate", learningRate);
		conf.setFloat("model.eps", eps);
		conf.set("model.name", modelName);
		conf.set("model.path",outputPath);
		conf.setFloat("model.reg", d);
		conf.set("net.master", master);
		conf.setInt("net.masterport", masterPort);
		conf.setJobName("Logistic Regression");
		
		conf.setOutputKeyClass(NullWritable.class);
		conf.setOutputValueClass(NullWritable.class);
		conf.setMapperClass(LRMapper.class);
		conf.setNumReduceTasks(0);
		
		conf.setInputFormat(TextInputFormat.class);
		conf.setOutputFormat(TextOutputFormat.class);
		
		FileInputFormat.setInputPaths(conf, new Path(inputPath));
		FileOutputFormat.setOutputPath(conf, new Path(outputPath));
		
		JobClient.runJob(conf);
	}

}
